Node / ReticulumProjects / MeshChatX.git / files / meshchatx / src / backend / async_utils.py

Displaying Raw • Download

T8b949e# SPDX-License-Identifier: 0BSD AND MIT

Tff7b72import T7ee787asyncio

Tff7b72import T7ee787threading

Tff7b72from T7ee787collectionsT7ee787.T7ee787abc Tff7b72import Te6edf3Coroutine

Tff7b72from T7ee787typing Tff7b72import Te6edf3AnyTb4b4b4, Te6edf3ClassVar

Tff7b72class T56d364AsyncUtilsTb4b4b4:

Te6edf3main_loopTb4b4b4: Te6edf3asyncioTff7b72.Td2a8ffAbstractEventLoop Tff7b72| Tff7b72None Tff7b72= Tff7b72None

Te6edf3_pending_futuresTb4b4b4: Te6edf3ClassVarTb4b4b4[Tffa657listTb4b4b4[Te6edf3AnyTb4b4b4]Tb4b4b4] Tff7b72= Tb4b4b4[Tb4b4b4]

Te6edf3_pending_coroutinesTb4b4b4: Te6edf3ClassVarTb4b4b4[Tffa657listTb4b4b4[Te6edf3AnyTb4b4b4]Tb4b4b4] Tff7b72= Tb4b4b4[Tb4b4b4]

Te6edf3_futures_lock Tff7b72= Te6edf3threadingTff7b72.Td2a8ffLockTb4b4b4(Tb4b4b4)

Te6edf3_FUTURES_SWEEP_THRESHOLD Tff7b72= T79c0ff32

Te6edf3_COROUTINES_MAX Tff7b72= T79c0ff256

Tf0883e@staticmethod

Tff7b72def Td2a8ffset_main_loopTb4b4b4(Te6edf3loopTb4b4b4: Te6edf3asyncioTff7b72.Td2a8ffAbstractEventLoopTb4b4b4)Tb4b4b4:

Te6edf3AsyncUtilsTff7b72.Td2a8ffmain_loop Tff7b72= Te6edf3loop

Tff7b72for Te6edf3coro Tff7b72in Te6edf3AsyncUtilsTff7b72.Td2a8ff_pending_coroutinesTb4b4b4:

Te6edf3AsyncUtilsTff7b72.Td2a8ffrun_asyncTb4b4b4(Te6edf3coroTb4b4b4)

Te6edf3AsyncUtilsTff7b72.Td2a8ff_pending_coroutinesTff7b72.Td2a8ffclearTb4b4b4(Tb4b4b4)

Tf0883e@staticmethod

Tff7b72def Td2a8ffrun_asyncTb4b4b4(Te6edf3coroutineTb4b4b4: Te6edf3CoroutineTb4b4b4)Tb4b4b4:

T8b949e"""Schedule *coroutine* on the main event loop from any thread.

Returned futures are tracked so they (and the closures they reference)

can be garbage-collected promptly once finished.

"""

Tff7b72if Te6edf3AsyncUtilsTff7b72.Td2a8ffmain_loop Tff7b72and Te6edf3AsyncUtilsTff7b72.Td2a8ffmain_loopTff7b72.Td2a8ffis_runningTb4b4b4(Tb4b4b4)Tb4b4b4:

Te6edf3future Tff7b72= Te6edf3asyncioTff7b72.Td2a8ffrun_coroutine_threadsafeTb4b4b4(

Te6edf3coroutineTb4b4b4,

Te6edf3AsyncUtilsTff7b72.Td2a8ffmain_loopTb4b4b4,

Tb4b4b4)

Tff7b72with Te6edf3AsyncUtilsTff7b72.Td2a8ff_futures_lockTb4b4b4:

Te6edf3AsyncUtilsTff7b72.Td2a8ff_pending_futuresTff7b72.Td2a8ffappendTb4b4b4(Te6edf3futureTb4b4b4)

Tff7b72if Tb4b4b4(

Tffa657lenTb4b4b4(Te6edf3AsyncUtilsTff7b72.Td2a8ff_pending_futuresTb4b4b4)

Tff7b72>Tff7b72= Te6edf3AsyncUtilsTff7b72.Td2a8ff_FUTURES_SWEEP_THRESHOLD

Tb4b4b4)Tb4b4b4:

Te6edf3AsyncUtilsTff7b72.Td2a8ff_pending_futures Tff7b72= Tb4b4b4[

Te6edf3f Tff7b72for Te6edf3f Tff7b72in Te6edf3AsyncUtilsTff7b72.Td2a8ff_pending_futures Tff7b72if Tff7b72not Te6edf3fTff7b72.Td2a8ffdoneTb4b4b4(Tb4b4b4)

Tb4b4b4]

Tff7b72return

T8b949e# If the loop isn't available yet (or has stopped), buffer the coroutine

T8b949e# but cap the backlog so we don't leak memory indefinitely.

Te6edf3AsyncUtilsTff7b72.Td2a8ff_pending_coroutinesTff7b72.Td2a8ffappendTb4b4b4(Te6edf3coroutineTb4b4b4)

Tff7b72if Tffa657lenTb4b4b4(Te6edf3AsyncUtilsTff7b72.Td2a8ff_pending_coroutinesTb4b4b4) Tff7b72> Te6edf3AsyncUtilsTff7b72.Td2a8ff_COROUTINES_MAXTb4b4b4:

Te6edf3dropped Tff7b72= Tffa657lenTb4b4b4(Te6edf3AsyncUtilsTff7b72.Td2a8ff_pending_coroutinesTb4b4b4) Tff7b72- Te6edf3AsyncUtilsTff7b72.Td2a8ff_COROUTINES_MAX

Te6edf3AsyncUtilsTff7b72.Td2a8ff_pending_coroutines Tff7b72= Te6edf3AsyncUtilsTff7b72.Td2a8ff_pending_coroutinesTb4b4b4[

Tff7b72-Te6edf3AsyncUtilsTff7b72.Td2a8ff_COROUTINES_MAX Tb4b4b4:

Tb4b4b4]

Tff7b72import T7ee787logging

Te6edf3loggingTff7b72.Td2a8ffgetLoggerTb4b4b4(Ta5d6ff"Ta5d6ffmeshchatx.asyncTa5d6ff"Tb4b4b4)Tff7b72.Td2a8ffwarningTb4b4b4(

Ta5d6ff"Ta5d6ffDropped Tffd700%dTa5d6ff buffered coroutine(s) because the event loop is not runningTa5d6ff"Tb4b4b4,

Te6edf3droppedTb4b4b4,

Tb4b4b4)